package com.atguigu.PracticalProject.app.dwd.db;


import com.atguigu.PracticalProject.app.BaseSqlApp;

import com.atguigu.PracticalProject.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
//这个表在ads层需要关联 user_info test_paper course_info三个维度表
public class DwdStudyExamDetail extends BaseSqlApp {
    public static void main(String[] args) {
        new DwdStudyExamDetail().init(5000, 2, "DwdStudyExamQuestionDetail");
    }


    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        readOdsDb(tEnv, "DwdStudyExamQuestionDetail");
        Table testExam = tEnv.sqlQuery("" +
                " select " +
                " `data`['user_id'] user_id," +
                " `data`['id'] id," +
                " `data`['paper_id'] paper_id," +
                " cast(`data`['score'] as double) score," +
                " cast(`data`['duration_sec'] as bigint) duration_sec," +
                " `data`['create_time'] create_time," +
                " UNIX_TIMESTAMP(`data`['create_time']) ts" +
                " from ods_db" +
                " where `database`='edu' and `table` = 'test_exam'" +
                " and `type` <> 'bootstrap-start'" +
                " and `type` <> 'bootstrap-complete'" +
                " ");
//        testExam.execute().print();
        tEnv.executeSql("" +
                " create table exam_question(" +
                " user_id string," +
                " id string," +
                " paper_id string," +
                " score double," +
                " duration_sec bigint," +
                " create_time string," +
                " ts bigint" +
                " )" + SQLUtil.getKafkaSink("dwd_course_paper_exam")
        );
        testExam.executeInsert("exam_question");
    }
}
